/*
* Copyright (c) 2008 by Kirill Kolodyazhniy.
* See the file "license.terms" for information on usage and redistribution.
*/

#include "threadpool.h"
#include "functors.h"
#include <algorithm>

#ifdef _DEBUG
#ifdef _MSC_VER
   #define new new( _CLIENT_BLOCK, __FILE__, __LINE__)
#endif
#endif // _DEBUG

namespace Engine
{
	using namespace std;
	/*******************************************************************************************/
	bool TaskExecutor::Task(unsigned long*)
	{
		if(task != 0)
		{
			task->Run();
			task->SetRuning(false);
		}

		return true;
	}
	/**********************************************************************************************************/
	template<> ThreadPool* Singleton<ThreadPool>::m_singleton = 0;
	/*******************************************************************************************/
	ThreadPool::ThreadPool(unsigned int num, unsigned int maxNum)
		:m_threadPool(maxNum)
		,m_freeThreads(m_threadPool.NewInstance())
		,m_threadsCount(0)
	{
		if(num > maxNum)
			num = maxNum;

		SetThreadsNum(num);
	}
	/*******************************************************************************************/
	void ThreadPool::FreeThreadsQueue(unsigned int num)
	{
		if(num >= m_threads.size())
			return;

		//stop all threads
		ThreadList::iterator i = m_threads.begin(), e = m_threads.end();
		for(;i != e; ++i)
		{
			(*i)->RequestStop();
			(*i)->WaitUntilDone();
		}

		LockFree::node<ThreadRawPtr>* pNode = 0;
		while( (pNode = m_freeThreads.Remove()) != 0)
			m_threadPool.FreeInstance(pNode);

		m_threads.resize(num);
		for_each(m_threads.begin(), m_threads.end(), m_fn(&Thread<TaskExecutor>::Run));
	}
	/*******************************************************************************************/
	ThreadPool::~ThreadPool()
	{
		FreeThreadsQueue(0);
		if(LockFree::node<ThreadRawPtr>* pNode = m_freeThreads.GetHead())
			m_threadPool.FreeInstance(pNode);
	}
	/*******************************************************************************************/
	ThreadPool::ThreadRawPtr ThreadPool::ReserveThread()
	{
		ThreadRawPtr th = 0;
		if(LockFree::node<ThreadRawPtr>* pNode = m_freeThreads.Remove())
		{
			th = pNode->value;
			m_threadPool.FreeInstance(pNode);
		}
		return th;
	}
	/*******************************************************************************************/
	void ThreadPool::ReturnThread(Thread<TaskExecutor>* th)
	{
		if(th != NULL)
		{
			th->GetThreadObject()->task = 0;
			LockFree::node<ThreadRawPtr>* pNode = m_threadPool.NewInstance();
			pNode->value = th;
			m_freeThreads.Add(pNode);
			return;
		}
		throw std::invalid_argument("Not initialized thread");
	}
	/*******************************************************************************************/
	void ThreadPool::SetThreadsNum(unsigned int num)
	{
		if(m_threadPool.GetSize() < num)
			num = m_threadPool.GetSize();

		if(m_threadsCount > num)
		{
			FreeThreadsQueue(num);
		}
		else if(num < m_threadPool.GetSize() && num > m_threadsCount)
		{
			unsigned int len = num;
			for(unsigned int i = 0; i < len; ++i)
			{
				m_threads.push_back(ThreadPtr(new Thread<TaskExecutor>(0,INFINITE)));
				m_threads.back()->WaitWorkSignal.Connect<ImmediateExecute>(this,&ThreadPool::ReturnThread);
				m_threads.back()->Run();
			}
		}
		m_threadsCount = num;
	}
	/*******************************************************************************************/
	unsigned int ThreadPool::GetThreadsNum()const
	{
		return m_threadsCount;
	}
	/*******************************************************************************************/
	void ThreadPool::Run(RefPtr<BaseTask> task)
	{
		task->SetRuning(true);

		ThreadRawPtr th = ReserveThread();
		if(th != NULL)
		{
			th->GetThreadObject()->task = task.GetPointer();
			th->DoWork();
			return;
		}

		task->Run();
		task->SetRuning(false);
	}
    /*******************************************************************************************/
}


